package com.bw.ads;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/*
(1,192.168.0.1,fail,1558430842) A
(1,192.168.0.2,fail,1558430843) B
1,192.168.0.2,fail,1558430844
1,192.168.0.3,fail,1558430850
1,192.168.0.3,fail,1558430851
2,192.168.10.10,fail,1558430851
2,192.168.10.10,fail,1558430858
2,192.168.10.10,fail,1558430864
2,192.168.10.10,fail,1558430878
 */
public class FlinkTM5_T_01_弃用 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment fsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(fsEnv);

        fsTableEnv.executeSql("create table dwd_order_info(" +
                "    id STRING," +
                "    user_id STRING," +
                "    order_status STRING," +
                "    create_time STRING, " +
                "    rowtime AS TO_TIMESTAMP(create_time)," +
                "    WATERMARK FOR rowtime AS rowtime" +
                "    ) with (" +
                "  'connector' = 'kafka',  " +
                "  'topic' = 'order_info_yk1'," +
                "  'properties.bootstrap.servers' = 'hadoop102:9092', " +
                "  'properties.group.id' = 'flinktm5_t_01'," +
                "  'format' = 'json'," +
                "  'scan.startup.mode' = 'earliest-offset')"
        );

//        fsTableEnv.sqlQuery("select * from dwd_order_info").execute().print();
//
        fsTableEnv.sqlQuery("SELECT T.*\n" +
                "            FROM t_cep\n" +
                "            MATCH_RECOGNIZE (\n" +
                "              PARTITION BY uid\n" +
                "              ORDER BY times\n" +
                "              MEASURES\n" +
                "                A.user_id as auid,A.order_status as astatus ,A.create_time as atime\n" +
                "              PATTERN (A{2}) WITHIN INTERVAL '5' MINUTE\n" +
                "              DEFINE\n" +
                "                A AS status = '1004'" +
                "            ) AS T").execute().print();
//

        //fsEnv.execute();

    }
}
